package com.atguigu.realtime.app.dwd;

import com.atguigu.realtime.app.BaseSQLApp;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dwd_CourseOrderDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_CourseOrderDetail().init(
                3004,
                2,
                "Dwd_CourseOrderDetail"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tabEnv) {
        readOdsDb(tabEnv, "Dwd_CourseOrderDetail");
        readBaseDic(tabEnv);

        //TODO 过滤订单表
        Table orderInfo = tabEnv.sqlQuery("select " +
                "data[`id`] id, " +
                "data[`user_id`] user_id " +
                "from topic_db " +
                "where `database`='edu' " +
                "and `table`='order_info' " +
                "and `type`='insert' ");
        tabEnv.createTemporaryView("order_info", orderInfo);

        //TODO 过滤订单详情表
        Table orderDetail = tabEnv.sqlQuery("select " +
                "data[`id`] id, " +
                "data[`order_id`] order_id, " +
                "data[`course_id`] course_id, " +
                "data[`course_name`] course_name, " +
                "data[`user_id`] user_id, " +
                "" +
                "from topic_db " +
                "where `database`='edu' " +
                "and `table`='order_detail' " +
                "and `type`='insert' ");
        tabEnv.createTemporaryView("order_detail", orderDetail);

        //TODO
    }
}
